package com.auditlog.listener;

import cn.hutool.core.thread.NamedThreadFactory;
import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.EventBus;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
public class AuditLogEventBus {

    private static final AuditLogEventBus INSTANCE = new AuditLogEventBus();

    private static final int QUEUE_SIZE = 2048;

    private final EventBus eventBus;

    private final String identifier = "auditLog-";

    private AuditLogEventBus() {
        int availableProcessors = Runtime.getRuntime().availableProcessors();
        ExecutorService executor = new ThreadPoolExecutor(1, availableProcessors / 2 == 0 ? 1 : availableProcessors / 2,
                Integer.MAX_VALUE, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(QUEUE_SIZE), new NamedThreadFactory(identifier, true), (r, e) -> {
            log.warn("队列数达到上线：{}", e.getQueue().size());
        });
        eventBus = new AsyncEventBus(identifier, executor);
        eventBus.register(new TableChangeEventSubscriber());
        eventBus.register(new ContextChangeEventSubscriber());
    }

    public static AuditLogEventBus getInstance() {
        return INSTANCE;
    }

    public void register(Object subscriber) {
        INSTANCE.eventBus.register(subscriber);
    }

    public void unregister(Object subscriber) {
        INSTANCE.eventBus.unregister(subscriber);
    }

    public void publish(Object event) {
        INSTANCE.eventBus.post(event);
    }

}
